package timeandwindow;

/*
*案例需求：统计10秒内的用户点击次数（pv）
*
* 增量聚合函数：
*
*   ReduceFunction：两两归约，上一次的聚合结果和本次的数据进行聚合。输入类型和输出类型保持一致。
*
* */

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import util.SourceUtil;

import java.time.Duration;

public class Flink06_ReduceFunction {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);


         SingleOutputStreamOperator<Event> ds = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource")
                 .assignTimestampsAndWatermarks(
                         WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                 .withTimestampAssigner(
                                         (event, timestamp) -> event.getTs()
                                 )
                 );
         ds.print("INPUT");

         //todo 1.需求统计10秒内，用户点击次数（pv）
         //思路：非按键事件时间滚动窗口

         //1)转化数据为Tuple1
         ds.map(
                 event -> Tuple1.of(1L)
         ).returns(
                 Types.TUPLE(Types.LONG)
         ).windowAll(
                 TumblingEventTimeWindows.of(Time.seconds(10))
         ).reduce(
                 new ReduceFunction<Tuple1<Long>>() {
                     @Override
                     public Tuple1<Long> reduce(Tuple1<Long> v1, Tuple1<Long> v2) throws Exception {
                         return Tuple1.of(v1.f0 + v2.f0);
                     }
                 }
         ).print("pv");

         //2)直接求和


         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }

}
